package com.atguigu.edu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.app.BaseApp;
import com.atguigu.edu.bean.ExamCount;
import com.atguigu.edu.bean.TradeProvinceOrderWindow;
import com.atguigu.edu.bean.TrafficPageViewBean;
import com.atguigu.edu.common.Constant;
import com.atguigu.edu.util.AtguiguUtil;
import com.atguigu.edu.util.DimUtil;
import com.atguigu.edu.util.FlinkSinkUtil;
import com.atguigu.edu.util.JdbcUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.time.Duration;

public class Dws_12_interaction_exam_count extends BaseApp {
    public static void main(String[] args) {
        new Dws_12_interaction_exam_count().init(
                4012,
                3,
                "DwsApp_10_interaction_exam_user_count",
                Constant.TOPIC_DWD_TEST_EXAM
        );
    }
    @Override
    public void handle(StreamExecutionEnvironment ev, DataStreamSource<String> stream) {
        // 1. 解析成 pojo 类型
        SingleOutputStreamOperator<ExamCount> beanStream = parseToPojo(stream);
        // 2. 开窗聚和
        SingleOutputStreamOperator<ExamCount> streamWithoutDims = windowAndAgg(beanStream);
//        streamWithoutDims.print();
        // 3.补充维度信息
        SingleOutputStreamOperator<ExamCount> resultStream = joinDims(streamWithoutDims);
//        resultStream.print();
        // 4.写出到ck
        writeToClickHouse(resultStream);
    }

    private void writeToClickHouse(SingleOutputStreamOperator<ExamCount> resultStream) {
        resultStream.addSink(FlinkSinkUtil.getClickHouseSink("dws_interaction_exam_count",ExamCount.class));
    }

    private SingleOutputStreamOperator<ExamCount> joinDims(SingleOutputStreamOperator<ExamCount> streamWithoutDims) {
        return streamWithoutDims
                .map(new RichMapFunction<ExamCount, ExamCount>() {

                    private Connection conn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        conn = JdbcUtil.getPhoenixConnection();
                    }

                    @Override
                    public ExamCount map(ExamCount bean) throws Exception {
                        // 1. test_paper
                        JSONObject skuInfo = DimUtil.readDimFromPhoenix(conn, "dim_test_paper", bean.getPaperId());
                        bean.setPaperTitle(skuInfo.getString("PAPER_TITLE"));
                        bean.setCourseId(skuInfo.getString("COURSE_ID"));


                        // 2. course_info
                        JSONObject spuInfo = DimUtil.readDimFromPhoenix(conn, "dim_course_info", bean.getCourseId());
                        bean.setCourseName(spuInfo.getString("COURSE_NAME"));

                        return bean;
                    }
                });
    }

    private SingleOutputStreamOperator<ExamCount> windowAndAgg(SingleOutputStreamOperator<ExamCount> beanStream) {
        WatermarkStrategy<ExamCount> strategy = WatermarkStrategy.<ExamCount>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner((s, l) -> s.getTs())
                .withIdleness(Duration.ofSeconds(60));
        return beanStream.assignTimestampsAndWatermarks(strategy)
                .keyBy(t -> t.getPaperId() + t.getScore())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<ExamCount>() {
                    @Override
                    public ExamCount reduce(ExamCount value1, ExamCount value2) throws Exception {
                        value1.setUvCt(value2.getUvCt() + value1.getUvCt());
                        value1.setScoreCt(value2.getScoreCt() + value1.getScoreCt());
                        value1.setDurationSecCt(value2.getDurationSecCt() + value1.getDurationSecCt());
                        return value1;
                    }
                }, new ProcessWindowFunction<ExamCount, ExamCount, String, TimeWindow>() {
                    @Override
                    public void process(String s,
                                        Context context,
                                        Iterable<ExamCount> elements,
                                        Collector<ExamCount> out) throws Exception {
                        ExamCount bean = elements.iterator().next();
                        bean.setStt(AtguiguUtil.tsToDateTime(context.window().getStart()));
                        bean.setEdt(AtguiguUtil.tsToDateTime(context.window().getEnd()));
                        bean.setTs(System.currentTimeMillis());
                        out.collect(bean);
                    }
                });

    }

    private SingleOutputStreamOperator<ExamCount> parseToPojo(DataStreamSource<String> stream) {
        return stream.map(JSON::parseObject)
                .map(new MapFunction<JSONObject, ExamCount>() {
                    @Override
                    public ExamCount map(JSONObject obj) throws Exception {
                        Double score = obj.getDouble("score");
                        String scoreFlag = "";
                        if (score > 90){
                            scoreFlag = "[90,100)";
                        } else if (score > 80){
                            scoreFlag = "[80,90)";
                        } else if (score > 70){
                            scoreFlag = "[70,80)";
                        } else if (score > 69){
                            scoreFlag = "[60,70)";
                        } else {
                            scoreFlag = "[0,60)";
                        }
                        return ExamCount.builder()
                                .paperId(obj.getString("paper_id"))
                                .score(scoreFlag)
                                .uvCt(1L)
                                .scoreCt(score)
                                .durationSecCt(obj.getLong("duration_sec"))
                                .ts(obj.getLong("create_time"))
                                .build();
                    }
                });
    }
}
